//
// Programa que calcula una norma para cada una de las     |
// filas de una matriz bidimensional representada mediante |
// un formato de almacenamiento comprimido.                |
//
// Comando de compilacion: 
//                         gcc -o norma practica2.c -lm
//
#include <assert.h>
#include <limits.h>
#include <math.h>
#include <stdlib.h>
#include <stdio.h>
#include <sys/time.h>
#include "pvm3.h"



/* muestra un error y finaliza el programa */
void error(char * error)
{
	pvm_perror(error);
	pvm_exit();
	exit(0);    
}

/**************************************************************************************************
 * FUNCIONES DE ENVIO
 **************************************************************************************************/	 
/* envia un double de un proceso a otro */
void enviar_ndouble(double *numero, int mytid,int tid, int msgtag, int n)
{

	int count  = 1;
	int stride = 1;
	int i;

	if( pvm_initsend( PvmDataDefault ) < 0 )
		error("ERROR enviar_double(): pvm_initsend \n");
	for(i=0; i<n; i++){
		if( pvm_pkdouble( (&numero)[i], count, stride ) < 0 ) 
			error ("ERROR enviar_double(): pvm_pkdouble \n");
	}
	if( pvm_send( tid, msgtag) < 0 ) 
		error ("ERROR enviar_double(): pvm_send \n");

	printf("%d:DOUBLE[%d] message token[0]='%f' sent to %d\n",mytid,n,numero[0],tid);
}


/* recibe un double desde un proceso a otro */
void recibir_double(double *numero[], int mytid, int tid, int msgtag, int n)
{

	int count  = 1;
	int stride = 1;	
	int i = 0;

	if( pvm_recv( tid, msgtag ) < 0 ) 
		error ("ERROR recibir_double(): pvm_recv \n");
	for(i=0; i<n; i++){
		if( pvm_upkdouble(numero[i], count, stride ) < 0 ) 
			error ("ERROR recibir_double(): pvm_upkdouble \n");
	}

	printf("%d:DOUBLE[%d] message token[0]='%f' received from %d\n",mytid,n,(*numero)[0],tid);
}

/* envia un int de un proceso a otro */
void enviar_int(int *numero, int mytid,int tid, int msgtag){
	
	int count  = 1;
	int stride = 1;
    
    if( pvm_initsend( PvmDataDefault ) < 0 ) 
        error ("ERROR enviar_int(): pvm_initsend \n");
    if( pvm_pkint( numero, count, stride ) < 0 ) 
        error ("ERROR enviar_int(): pvm_pkdouble \n");
    if( pvm_send( tid, msgtag) < 0 ) 
        error ("ERROR enviar_int(): pvm_send \n");

    printf("%d:INT message token='%d' sent to %d\n",mytid,*numero,tid);
}


/* recibe un int de otro proceso */
void recibir_int(int *numero, int mytid, int tid, int msgtag){

	int count  = 1;
	int stride = 1;	
		
        printf("1  %d\n", mytid);
        fflush(NULL);
    if( pvm_recv( tid, msgtag ) < 0 ) 
        error ("ERROR recibir_int(): pvm_recv \n");
        printf("2  %d\n", mytid);
        fflush(NULL);        
    if( pvm_upkint(numero, count, stride ) < 0 ) 
        error ("ERROR recibir_int(): pvm_upkdouble \n");

    printf("%d:INT message token='%d' received from %d\n",mytid,*numero,tid);
}



/*************************************************************************************************/
	     
int main( int argc, char *argv[]) {

        int ptid;
        int mytid;
        int hijos;
	int *tids;
        int i=0;
        
	int envio = 1;
	int recibido;   
        
	mytid = pvm_mytid();	
	ptid = pvm_parent();	
	
	if(ptid < 0){
	        if (argc != 2) {
		        printf ("Uso: NHIJOS \n");
		        exit(0);
	        }
	        hijos = atoi(argv[1]);
	        /* crea procesos hijo */
	        tids =(int *)malloc(sizeof(int) * hijos);
		(tids)[0] = mytid;	
	        if( (pvm_spawn("prueba", (char**)0, 0, "", hijos-1, &(tids)[1])) < hijos-1)
		        error ("ERROR configure_parallel_program(): Not enough processes created \n");	
	        for(i=0; i<hijos; i++)		         
                        printf("TID [%d]: %d \n",i,tids[i]);
                for(i=1; i<hijos; i++)		         
                        enviar_int(&envio, tids[0], tids[i], 1);	               
	}else{
		recibir_int(&recibido, mytid, ptid, 1);	
	}
	
        

	printf("TIDS: %d %d\n",mytid, ptid);
	//printf("TIDS: %d %d, FILAS %d, COL %d, METODO %d, PROCESOS %d\n", mytid, ptid, nfilas, ncols, metodo, nproc);
	// Rutina de inicializacion ejecutada solamente por el proceso "padre"

	return 0;
}
